package com.atguigu.flink.window.agg;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2022/12/16
 *
 *      求过去每3个传感器中，vc的和
 *          reduce： 两两聚合
 *
 *
 *          按照聚合方式来分类:
 *              滚动聚合（和不开窗的聚合方式一样）:  来一条数据，就聚合一次。
 *                          reduce,aggretate,sum,max,maxBy,min,minBy
 *
 *                          特例: 基于元素个数的滑动窗口，以上算子变为非滚动聚合。
 *
 *              非滚动聚合:   process,apply
 *                              只有触发窗口运算时，才会进行计算。
 *                              把窗口的元素攒齐了，才计算。
 *
 *          容易混淆:       process 不开窗计算，是滚动聚合
 *                          开窗后使用process，是非滚动聚合
 *
 *     -------------------------------
 *      基于元素个数的滑动窗口:
 *               .countWindow(5,3)
 *
 *      s1,100,1
 *     s1,100,2
 *      s1,100,3
 *
 *      窗口中:  [ s1,100,1,  s1,100,2 ,  s1,100,3]
 *          触发运算: WaterSensor(id=s1, ts=100, vc=6)
 *
 *       修改窗口:  [ s1,100,1,  s1,100,3 ,  s1,100,6]
 *
 *     -----------------
 *      又来了
 *      s1,100,4
 *      s1,100,5
 *      s1,100,6
 *
 *       [ s1,100,3 ,  s1,100,6 ， s1,100,4 ，s1,100,5， s1,100,6]
 *
 *       解决：使用process聚合！
 *
 *
 */
public class Demo6_Reduce
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                    .socketTextStream("hadoop103", 8888)
                    .map(new WaterSensorMapFunction())
                    .keyBy(WaterSensor::getId)
                    // reduce是滚动聚合
                    //.countWindow(3)
                    // reduce变为了非滚动聚合
                    .countWindow(5,3)
                    .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>()
                    {
                        @Override
                        public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                            int result = StreamSupport.stream(elements.spliterator(), true)
                                                   .mapToInt(WaterSensor::getVc)
                                                   .sum();

                            out.collect("vc之和:"+result);

                        }
                    })
                   /* .reduce(new ReduceFunction<WaterSensor>()
                   {
                       @Override
                       public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                           System.out.println("Demo6_Reduce.reduce");
                           //每次输出最后一个元素的其他属性+vc和
                           value2.setVc(value1.getVc() + value2.getVc());
                           return value2;
                       }
                   })*/
                    .print();

                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
